/*
 * @Description: 
 * @Version: 1.0
 * @Author: Symbol_Yang
 * @Date: 2022-06-07 10:07:16
 * @LastEditors: Symbol_Yang
 * @LastEditTime: 2023-03-15 16:01:19
 */

const mqtt = require("mqtt");
const config = require('config')
const { EquipmentLogModel, EquipmentModel, EnterpriseModel, ShortMsgSendConfigModel, ShortMsgLogModel, ShortMsgModel, CabinetModel } = require('../models/index')
const cache = require('memory-cache')
const fs = require("fs")
const sendTxMsg = require("../commons/send_msg")
const ObjectId = require("mongoose").Types.ObjectId;
const iconv = require("iconv-lite")

// 设备数据缓存 -- 优化数据处理性能；
let equipmentCache = {};
// 柜数据缓存
let cabinetCache = {};

// 发送短信 同柜号下的设备发生异常时 
let warnCabinetMsg = {
    /**
     * cabinet_id: {
     *    refreshTime: Date.now(),
     *    count: 0,
     *    isFirst: false,
     * }
    */
}

// 发送信息配置
let msgCongigDataList = [];

// 企业手机号对应
let phone2EnterpriseMap = {};

// 待接入设备缓存记录
let unRelationEquipCache = {
    // code_1: {status: true, createAt: new Date()} 
}
// 定时器
let cacheClearTimer = {};

// 设备读数计时器
let equRealNumTimer = {
    // [equCode]: null;
}


class MqttServices {
    constructor() {
        const { host, port, clientId } = config.get('Customer.mqttConfig')

        this.clientId = `${clientId}-${Math.random().toString(16).slice(3)}`
        // this.clientId = `mqtt_node_dev_serve`
        // this.clientId = "mqtt_production_serve"
        this.client = null;
    }

    // mqttTopic = "atk/pub"
    mqttTopic = "test/send"

    machineTypeEnum = {
        "1": "G",
        "11": "D",
        "21": "T"
    }


    // 获取设备数据 (缓存)
    async getEquipmentData() {
        console.log("start update equipment data")
        let eqiDataList = await EquipmentModel.find({ status: 1 });
        equipmentCache = eqiDataList.reduce((a, b) => {
            // 兼容新的芯片唯一值
            a[b.wafer_uid || b.code] = b;
            return a;
        }, {});

        this.updateUnRealtionData();
    }

    // 获取柜数据
    async getCabinetData() {
        console.log("start update cabinet data");
        let cabinetDataList = await CabinetModel.find({ status: 1 }, "_id name").lean();
        // let _cabinetCache = {};
        cabinetDataList.forEach(item => {
            cabinetCache[item._id.toString()] = item.name;
        })
    }

    // 更新未接入的数据缓存
    updateUnRealtionData() {
        Object.values(unRelationEquipCache).forEach(item => {
            let equipItem = equipmentCache[item.wafer_id || item.code]
            if (equipItem && equipItem.status == 1) {
                item.exist = true;
            } else {
                item.exist = false;
            }
        })
    }

    // 获取未接入系统的设备列表
    getUnRelationDataList() {
        return Object.values(unRelationEquipCache).filter(item => !item.exist)
    }

    // 消除设备响应缓存
    clearCacheRelationData(){
        unRelationEquipCache = {};
    }


    // 获取发送短信配置
    async initMsgConfigData(type = 'all') {

        if (type == 'all') {
            let cacheJsonData = fs.readFileSync("./cache.json");
            if (cacheJsonData) {
                try{
                    warnCabinetMsg = JSON.parse(cacheJsonData);
                }catch(e){
                    warnCabinetMsg = {};
                }
            } 
        }

        msgCongigDataList = await ShortMsgSendConfigModel.find().sort({ sort: 1 });

        let enterpriseDataList = await EnterpriseModel.find({ status: 1 }, "_id concat_phone send_msg_status").lean();
        phone2EnterpriseMap = enterpriseDataList.reduce((a, b) => Object.assign(a, { [b._id]: ({ phone: b.concat_phone, status: b.send_msg_status }) }), {});
    }

    enterpriseTopicData = new Map();
    // 获取企业的订阅主题数据
    async getEnterpriseTopic(){
        let eQuery = {
            status: 1,
            topic: {$exists: true, $ne: ""}
        };
        let enterpriseDataList = await EnterpriseModel.find(eQuery, "topic").lean();
        enterpriseDataList.forEach(item => {
            this.enterpriseTopicData.set(item._id.toString(), item.topic)
        })
    }

    async startMqtt() {
        await this.getEquipmentData();
        await this.getCabinetData();
        await this.getEnterpriseTopic()
        await this.initMsgConfigData("all");
        const { host, port, username, password } = config.get('Customer.mqttConfig')
        const connectUrl = `mqtt://${host}:${port}`
        let client = null;
        this.client = client = mqtt.connect(connectUrl, {
            clientId: this.clientId,
            clean: true,
            connectTimeout: 4000,
            username: username,
            password: password,
            reconnectPeriod: 1000
        });
        console.log("client", client.options.clientId)
        // 定义主题，接受所有的设备发出的信息
        client.on("connect", () => {
            console.log("mqtt connect success");
            // 开始订阅
            let subTopics = ["atk/pub", "sub/atk", "anzhitekPub", "anzhitekPubV2.0"]
            // let subTopics = ["anzhitekPub"]
            // let subTopics = ["test/send"]
            client.subscribe(subTopics, () => {
                console.log(`MQTT  订阅成功`)
            });

        })
        // 接收信息
        client.on("message", (subTopic, payload) => {
            if (["atk/pub", "sub/atk"].includes(subTopic)) {
                this.mqttMessage(payload)
            } else if (subTopic === "anzhitekPub") {
                // console.log(payload.toString())
                try {
                    this.mqttMessage_v2(JSON.parse(payload.toString()));
                } catch (error) {
                    console.log("解析异常", error)
                }
            } else if (subTopic === "anzhitekPubV2.0") { 
                try {
                    let payloadStr = payload.toString();
                    // 过滤非Json格式数据
                    if (!payloadStr.includes("{")) return
                    console.log("anzhitekPubV2.0", payloadStr)
                    if (payloadStr.length < 100) {
                        this.mqttMessage_v3(JSON.parse(payload.toString()));
                    }
                } catch (error) {
                    console.log("解析异常")
                    // console.log(error)
                }
            }
        })

        // 断开MQTT服务回调
        client.on("disconnect", () => {
        })
    }

    /**
     * ethyne 乙炔 C2H2
     * propane 丙烷 C3H8
     * oxygen 氧气 O2
     * propylene 丙烯 C3H6
     * */
    // 解析存在多种气体模式的数据
    N_2_TYPE = {
        3: "GAS",
        7: "O2",
        8: "temp"
    }
    mqttMessage_v3(jsonData) {
        let { uid, t, v, n, s} = jsonData;
        // 231108 -- 兼容 n参数 去除t的情况
        if(!t && n){
            t = this.N_2_TYPE[n] || n
        }
        // 231206 -- 过滤 n 为0 的情况
        if(n === 0 || n === "0"){
            return 
        }
        // 231108 --end
        if (t === 'TEMP') {
            t = "temp"
        }
        let machineType = t === "temp" ? "T" : "G";
        let realNum = v || 0;
        // 当类型为温度 | 氧气是 将数值除以 10
        if (t === 'temp' || t === "O2") {
            realNum /= 10
        }
        if (machineType === "G") {
            machineType += '-' + t
        }
        // 设备UID 加上 传输的类型
        uid = (uid + "_" + machineType);
        let equData = equipmentCache[uid];
        this.createEquipData(equData, realNum, machineType, uid, "", s);
    }

    // 新 - 订阅主题
    mqttMessage_v2(jsonData) {
        let { uid, temperature, gas, desc, status } = jsonData;
        // 数据累加
        status += 1;
        let realNum = temperature || gas || 0;
        let machineType = temperature === undefined ? "G" : "T";
        if (machineType == 'T' && realNum) {
            realNum /= 10
        }
        // 设备UID 加上 传输的类型
        uid = (uid + "_" + machineType);
        let equData = equipmentCache[uid];
        // 若设备已录入，判断当前设备状态
        if (equData && equData.machine_status != status) {
            equData.machine_status = status;
            equData.save();
        }
        this.createEquipData(equData, realNum, machineType, uid, desc);
    }

    // 数据接收
    mqttMessage(payload) {
        let resData = this.bufferAnalysis(payload);
        if (!resData.result) return;
        let machineCode = resData.data.code;
        let equData = equipmentCache[machineCode];
        let realNum = (resData.data.machine_type == "T" && resData.data.real_num > 0) ? (resData.data.real_num / 10).toFixed(1) : resData.data.real_num
        // 生成数据
        this.createEquipData(equData, realNum, resData.data.machine_type, machineCode)
    }

    /**
     * 生成日志数据
     * @param {object} equData：    设备对象
     * @param {number} realNum：    当前读数
     * @param {string} machineType：设备类型
     * @param {string} keyCode   唯一值
     * @param {string} desc:        备注数据
     * @param {string} equStatus  设备状态
     * */
    createEquipData(equData, realNum, machineType, keyCode, desc = '', equStatus) {
        // 存在才进行
        if (equData && equData.status == 1) {
            // 温度值需要进行 / 10
            let equLogData = {
                enterprise_id: equData.enterprise_id,
                cabinet_id: equData.cabinet_id,
                equipment_id: equData._id,
                equipment_code: equData.code,
                real_num: realNum,
                remarks: desc,
                num_status: (realNum > equData.cordon) ? 2 : 1
            }
            // console.log("equ log data",  equLogData)
                    
            if(equStatus !== undefined){
                equLogData.equStatus = equStatus;
            }
            this.forward2PC(equLogData);
            
            // return false;
            // 发送异常数据
            this.sendErrorData(equLogData, machineType[0])
            // 开发环境不走下列逻辑
            if (this.clientId.includes("dev")) return false;
            // 生产环境下直系那个以下逻辑
            EquipmentLogModel.create(equLogData);

            // 更新对应设备的读数情况
            this.updateEquRealNum(equLogData);

            // 设备出现异常时的操作？ 发送短信 并进行记录日志填写
            // 设备数据超过了警戒线时的操作
            if (realNum < 300) {
                // 预警设备类型字典
                let typeEnum = { "T": "1", "G": "2" };
                let curItem = msgCongigDataList.find(item => equLogData.real_num > item.warning_num && typeEnum[machineType[0]] == item.type)
                if (curItem && phone2EnterpriseMap[equData.enterprise_id].status != 3 && phone2EnterpriseMap[equData.enterprise_id].phone) {
                    this.beforeSendMsg(equLogData, curItem);
                }
            }

        } else {
            let isExistEquip = unRelationEquipCache[keyCode];
            if (!isExistEquip) {
                unRelationEquipCache[keyCode] = {
                    code: keyCode,
                    createAt: Date.now(),
                    exist: false,
                }
            }

        }
    }

    // TODO 二次转发(合集)
    forward2PC(equLogData) {
        // 过滤异常数据
        if (equLogData.real_num > 6000) return;
        // console.log("equ log data",  equLogData)
        /**
         * 条件：
         *      相同企业 -- 气体、温度
         *      相同柜体 -- 
         * 
        */
        let bufData = {
            eId: equLogData.enterprise_id,
            equId: equLogData.equipment_id,
            // 设备中止逻辑再走数据异常逻辑
            n_s: equLogData.equStatus ? 3 : equLogData.num_status,
            n: equLogData.real_num,
        }
        // 发送
        let sendTopic = "equ_log"
        // this.client.publish(sendTopic, new Buffer.from(JSON.stringify(bufData)), { qos: 0, retain: false }, () => {})

        // 单独发送给企业
        let sendEnterTopic = `equ_log_${equLogData.enterprise_id}`;
        let enterBufData = {
            cId: equLogData.cabinet_id,
            eId: equLogData.equipment_id,
            v: equLogData.real_num,
            s: equLogData.equStatus
        }
        this.client.publish(sendEnterTopic, new Buffer.from(JSON.stringify(enterBufData)), { qos: 0, retain: false }, () => {})
    }

    // 发送异常数据
    sendErrorData(equLogData, machineType) {
        let { real_num, num_status, cabinet_id, enterprise_id } = equLogData
        // 过滤异常数据
        if (real_num > 6000) return;
        if (num_status == 1) return;
        // console.log("send error data", equLogData)

        let cabName = cabinetCache[cabinet_id]
        let equTypeName = machineType === "T" ? '温度传感器' : "气体传感器"
        let buffer = `${cabName},${equTypeName}`
        buffer = iconv.encode(buffer, 'gbk');
        // 需对其发送方式做节流处理
        this.sendSustainErrorData(buffer, enterprise_id)
        /**
         * */
    }

    /**
     *发送持续异常数据
     * @memberof MqttServices
     * @param buffer 待发送数据
     * 需对其方法内做防抖处理
     */
    throttle = {};
    sendSustainErrorData(buffer, enterpriseId) {
        // let topic = "anzhitekALARM"
        let topic = this.enterpriseTopicData.get(enterpriseId.toString()) || "";
        if (this.throttle[topic]) return;
        // 警示3s 间隔时间为300ms
        let count = 0
        this.throttle[topic] = setInterval(() => {
            topic.split(',').forEach(topicItem => {
                topicItem && this.client.publish(topicItem, buffer, { qos: 0, retain: false }, (error) => {
                    if (error) {
                        console.error(error)
                    }
                })
            })
            if (++count == 15) {
                clearInterval(this.throttle[topic]);
                this.throttle[topic] = null
            }
        }, 300);

    }

    // TODO 更新设备读数
    updateEquRealNum(equLogData) {

        // 当设备读数大于1000时，判定为异常，不做设备读数更新
        if(equLogData.real_num > 1000) return;

        let equId = equLogData.equipment_id;
        let equIdCountKey = equId + "_count";
        if (!equRealNumTimer[equIdCountKey]) {
            equRealNumTimer[equIdCountKey] = 0;
        }
        
        let updatFn = async () => {
            equRealNumTimer[equIdCountKey] = 0;
            // console.log("update equ data", equId, equLogData.real_num)
            let setData = {
                real_num: +equLogData.real_num,
                real_time: new Date(),
            }
            if(equLogData.equStatus !== undefined){
                // 1: 正常；2：异常；3：中止
                setData.machine_status = equLogData.equStatus == "0" ? 1 : 3
            }
            await EquipmentModel.updateMany({ _id: equId }, {
                $set: setData
            })
        }
        equRealNumTimer[equIdCountKey]++;
        // 清除定时器
        clearTimeout(equRealNumTimer[equId]);
        // 当短时间内触发次数达到指定次数时，直接触发修改
        if (equRealNumTimer[equIdCountKey] == 10) {
            updatFn();
        } else {
            // 防抖性处理，防止数据刷新过于频繁；
            equRealNumTimer[equId] = setTimeout(() => {
                updatFn();
            }, 2000)
        }

    }

    // TODO字节解析
    bufferAnalysis(buffer) {

        if (buffer.length != 12) return { result: false, data: "数据格式不符" };
        let isFrameH = new Array(3).fill(1).every((_, i) => buffer[i].toString(16) == "aa");
        let isFrameF = new Array(3).fill(1).every((_, i) => buffer[11 - i].toString(16) == "bb");
        if (!(isFrameF && isFrameH)) return { result: false, data: "帧头或者帧尾不符" };;

        // 开始解析
        // 类型 1,2 为气体 （G） 3、为温度 （T）
        let machineType = ""
        switch (buffer[5].toString(16)) {
            case "1":
            case "2":
                machineType = "G";
                break;
            case "3":
                machineType = "T";
                break;
            default: ; break;
        }

        let machineCodeHex = 0x0;
        // 编号
        new Array(2).fill(1).forEach((item, index) => {
            machineCodeHex += this.fillZero(buffer[index + 3].toString(16))
        })

        // 读数
        let realNumHex = 0x0;
        new Array(2).fill(1).forEach((item, index) => {
            realNumHex += this.fillZero(buffer[index + 6].toString(16))
        })
        let resData = {
            machine_type: machineType,
            code: `${machineType}_${this.fillZero(Number(`0x${machineCodeHex}`), 4)}_${('0' + buffer[5])}`,
            real_num: Number(`0x${realNumHex}`),
            machine_status: buffer[8]
        }
        return { result: true, data: resData }

    }

    fillZero(num, pos = 2) {
        let zeroNum = new Array(pos).fill("0").join("");
        return (zeroNum + num).slice(0 - pos);
    }


    //  需求 - 温度在 28 和 30 需进行发送  或 超过预警线 
    beforeSendMsg(eLogData, confirmData) {

        // 是否存在记录 key规则= 柜Id + 短信发送阶段配置sort + 设备类型
        let key = eLogData.cabinet_id + "_" + confirmData.sort + "_" + confirmData.type;
        let tWarnCacheData = warnCabinetMsg[key]
        // 判断是否为首次触发
        // 1、越线即发送
        if (confirmData.send_setting == 1 && !tWarnCacheData) {
            warnCabinetMsg[key] = {
                refreshTime: Date.now(),
                count: 1,
                isNoFirst: true,
            }
            return this.sendMng(eLogData, confirmData)
        }

        // 2、1分钟内出现5次预警即发送
        if (confirmData.send_setting == 2) {
            // 首次定义
            if (!tWarnCacheData) {
                warnCabinetMsg[key] = {
                    refreshTime: Date.now(),
                    count: 1,
                }
            } else {
                // 检查是否超过了1分钟, 超过则重新进行计数
                if (((Data.now() - warnCabinetMsg[key].refreshTime) / 1000 / 60) > 1) {
                    warnCabinetMsg[key].count = 0;
                }

                warnCabinetMsg[key].refreshTime = Date.now();
                warnCabinetMsg[key].count++;
            }
            if (warnCabinetMsg[key].count == 5) {
                warnCabinetMsg[key].isNoFirst = true;
                return this.sendMng(eLogData, confirmData)
            }

        }

        // 当首次短信发送之后，在一定的间隔时间内 重新进行触发即发送
        tWarnCacheData = warnCabinetMsg[key]
        if (tWarnCacheData && tWarnCacheData.isNoFirst) {
            let diffMin = (Date.now() - tWarnCacheData.refreshTime) / 1000 / 60;
            // console.log("enter pause_time",diffMin, confirmData.pause_time )
            if (diffMin > confirmData.pause_time) {
                // 更新存储时间
                tWarnCacheData.refreshTime = Date.now();
                // 后一条件有效控制服务重启后，缓存时间问题
                if (diffMin < confirmData.pause_time + 1) {
                    return this.sendMng(eLogData, confirmData)
                }
            }
        }

        // 定时清除超时数据  
        clearTimeout(cacheClearTimer[key]);
        cacheClearTimer[key] = setTimeout(async () => {
            // 更新企业数据的发送短信状态
            let enterId = eLogData.enterprise_id;
            let tEnterpriseData = await EnterpriseModel.findById(enterId);
            if (tEnterpriseData && tEnterpriseData.send_msg_status != 3) {
                tEnterpriseData.send_msg_status = 1;
                await tEnterpriseData.save();
                // 更新当前环境下的企业集合对象
                phone2EnterpriseMap[enterId].status = 1;

            }
            delete warnCabinetMsg[key];
        }, confirmData.pause_time * 60 * 1000);


    }

    // 新增短信待发送日志
    async sendMng(eLogData, confirmData) {
        // 企业id 柜Id 企业手机号 发送内容  模板id
        let mLogData = {
            enterprise_id: eLogData.enterprise_id,
            cabinet_id: eLogData.cabinet_id,
            send_phone: phone2EnterpriseMap[eLogData.enterprise_id].phone,
            content: `警告值：${eLogData.real_num}`,
            short_msg_id: confirmData.short_msg_id,
        };
        console.log("log data", mLogData)
        // 生成后的日志数据
        let nLogData = await ShortMsgLogModel.create(mLogData);
        // 需要改企业配置状态
        let enterId = eLogData.enterprise_id;
        let tEnterpriseData = await EnterpriseModel.findById(enterId);
        // 企业数据存在 & 短信状态不为关闭状态
        if (tEnterpriseData && tEnterpriseData.send_msg_status != 3) {
            tEnterpriseData.send_msg_status = 2;
            await tEnterpriseData.save();
            let shortMsgData = await ShortMsgModel.findById(confirmData.short_msg_id).lean();
            if (mLogData.send_phone && shortMsgData) {
                // 开始发送短信
                let msgParams = {
                    PhoneNumberSet: ['+86' + mLogData.send_phone],
                    SmsSdkAppId: shortMsgData.appId,
                    SignName: shortMsgData.signName,
                    TemplateId: shortMsgData.templateId,
                    // TemplateParamSet: [eLogData.real_num],
                };
                if (shortMsgData.hasIncludeParams) {
                    msgParams.TemplateParamSet = [eLogData.real_num];
                }
                sendTxMsg(msgParams).then(res => {
                    nLogData.remarks = JSON.stringify(res);
                    // 正常发送
                    if (nLogData.remarks.includes("send success")) {
                        nLogData.send_status = 2;
                    } else { // 发送异常
                        nLogData.send_status = 3;
                    }
                    nLogData.save();
                }).catch(err => {
                    nLogData.remarks = JSON.stringify(err);
                    nLogData.save();
                })
            }

        }
    }

    // 缓存当前短信信息
    saveCacheData() {
        fs.writeFile("cache.json", JSON.stringify(warnCabinetMsg || "{}"), (err) => {
            if (err) {
                console.log("保存失败：" + err)
            }
        });
    }



    // 字节解析 弃用 2022.07.23
    bufferAnalysis_1(buffer) {

        if (buffer.length != 12) return { result: false, data: "数据格式不符" };
        let isFrameH = new Array(3).fill(1).every((_, i) => buffer[i].toString(16) == "aa");
        let isFrameF = new Array(3).fill(1).every((_, i) => buffer[11 - i].toString(16) == "bb");
        if (!(isFrameF && isFrameH)) return { result: false, data: "帧头或者帧尾不符" };;

        // 开始解析
        let machineType = this.machineTypeEnum[buffer[3].toString(16)]

        let machineCodeHex = 0x0;
        // 编号
        new Array(2).fill(1).forEach((item, index) => {
            machineCodeHex += buffer[index + 4].toString(16)
        })

        // 读数
        let realNumHex = 0x0;
        new Array(2).fill(1).forEach((item, index) => {
            realNumHex += buffer[index + 6].toString(16)
        })
        let resData = {
            machine_type: machineType,
            code: `${machineType}_${Number(`0x${+machineCodeHex}`)}`,
            real_num: Number(`0x${+realNumHex}`),
            machine_status: buffer[8]
        }
        return { result: true, data: resData }

    }

    mqttMessage1(subTopic, payload) {
        let resData = JSON.parse(payload.toString());

        let equData = equipmentCache[resData.clientId];
        // 存在才进行
        if (equData && equData.status == 1) {
            let equLogData = {
                enterprise_id: equData.enterprise_id,
                cabinet_id: equData.cabinet_id,
                equipment_id: equData._id,
                equipment_code: equData.code,
                real_num: resData.num,
                num_status: (+resData.num > equData.cordon) ? 2 : 1
            }
            EquipmentLogModel.create(equLogData)
        } else {
            let clientId = resData.clientId;
            let isExistEquip = unRelationEquipCache[resData.clientId];
            if (!isExistEquip) {
                unRelationEquipCache[clientId] = {
                    code: clientId,
                    createAt: Date.now(),
                    exist: false,
                }
            }

        }

    }

    closeMqtt() {
        console.log("MQTT disconnect");
        this.saveCacheData();
        this.client.end();
    }
}


module.exports = new MqttServices();




